package mq_kafka

import (
	"errors"
	"github.com/Shopify/sarama"
	"github.com/eapache/go-resiliency/breaker"
	"github.com/hashicorp/errwrap"
	"go.uber.org/zap"
	"log"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
)

type KafkaProducer struct {
	Name       string
	Hosts      []string
	Config     *sarama.Config
	Status     string
	Breaker    *breaker.Breaker
	Reconnect  chan bool
	StatusLock sync.Mutex
}

type KafkaMessage struct {
	Topic      string
	keyBytes   []byte
	ValueBytes []byte
}

type SyncKafkaProducer struct {
	KafkaProducer
	SyncProducer *sarama.SyncProducer
}

type AsyncKafkaProducer struct {
	KafkaProducer
	AsyncProducer *sarama.AsyncProducer
}

type StdLogger interface {
	Print(v ...interface{})
	Printf(format string, v ...interface{})
	Println(v ...interface{})
}

const (
	// KafkaProducerConnected 生产者已完成连接
	KafkaProducerConnected = "connected"
	// KafkaProducerDisConnected 生产者断开连接
	KafkaProducerDisConnected = "dis_connected"
	// KafkaProducerClosed 生产者已关闭
	KafkaProducerClosed = "closed"

	// DefaultKafkaSyncProducer 同步模式生产者的名称
	DefaultKafkaSyncProducer = "default_kafka_sync_producer"
	// DefaultKafkaAsyncProducer 异步模式生产者的名称
	DefaultKafkaAsyncProducer = "default_kafka_async_producer"
)

var (
	// ProducerTimeoutError 发送消息超时异常
	ProducerTimeoutError = errors.New("send message timeout")

	// kafkaSyncProducers 同步模式的kafka客户端连接列表
	kafkaSyncProducers = make(map[string]*SyncKafkaProducer)
	// kafkaAsyncProducers 异步模式的kafka客户端连接列表
	kafkaAsyncProducers = make(map[string]*AsyncKafkaProducer)

	// kafkaStdLogger 定义日志对象
	KafkaStdLogger StdLogger
)

func init() {
	KafkaStdLogger = log.New(os.Stdout, "[kafka]", log.LstdFlags|log.Lshortfile)
}

func KafkaMessageValueEncoder(value []byte) sarama.Encoder {
	return sarama.ByteEncoder(value)
}

func KafkaMessageValueStringEncoder(value string) sarama.Encoder {
	return sarama.StringEncoder(value)
}

// kafka生产者的默认配置
func getDefaultProducerConfig(ClientID string) (config *sarama.Config) {
	config = sarama.NewConfig()
	config.ClientID = ClientID
	config.Version = sarama.V2_0_0_0

	config.Net.DialTimeout = time.Second * 30
	config.Net.ReadTimeout = time.Second * 20
	config.Net.WriteTimeout = time.Second * 20

	config.Producer.Retry.Backoff = time.Millisecond * 500
	config.Producer.Retry.Max = 3

	config.Producer.Return.Successes = true
	config.Producer.Return.Errors = true

	//需要小于broker的 `message.max.bytes`配置，默认是1000000
	config.Producer.MaxMessageBytes = 1000000 * 2

	config.Producer.RequiredAcks = sarama.WaitForLocal
	//config.Producer.Partitioner = sarama.NewRandomPartitioner
	//config.Producer.Partitioner = sarama.NewRoundRobinPartitioner
	config.Producer.Partitioner = sarama.NewHashPartitioner
	//config.Producer.Partitioner = sarama.NewReferenceHashPartitioner

	// zstd 算法有着最高的压缩比，而在吞吐量上的表现只能说中规中矩，LZ4 > Snappy > zstd 和 GZIP
	//LZ4 具有最高的吞吐性能，压缩比zstd > LZ4 > GZIP > Snappy
	//综上LZ4性价比最高
	config.Producer.Compression = sarama.CompressionLZ4

	return
}

// InitSyncKafkaProducer 初始化同步生产者
func InitSyncKafkaProducer(name string, hosts []string, config *sarama.Config) error {
	syncProducer := &SyncKafkaProducer{}
	syncProducer.Name = name
	syncProducer.Hosts = hosts
	syncProducer.Status = KafkaProducerDisConnected
	if config == nil {
		config = getDefaultProducerConfig(name)
	}
	syncProducer.Config = config

	if producer, err := sarama.NewSyncProducer(hosts, config); err != nil {
		return errwrap.Get(err, "sarama.NewSyncProducer.error :"+name)
	} else {
		syncProducer.Breaker = breaker.New(3, 1, 2*time.Second)
		syncProducer.Reconnect = make(chan bool)
		syncProducer.SyncProducer = &producer
		syncProducer.Status = KafkaProducerConnected
		KafkaStdLogger.Printf("SyncKakfaProducer connected name %s", name)
	}
	go syncProducer.keepConnect()
	go syncProducer.check()

	kafkaSyncProducers[name] = syncProducer
	return nil
}

// InitAsyncKafkaProducer 初始化异步生产者
func InitAsyncKafkaProducer(name string, hosts []string, config *sarama.Config) error {
	asyncProducer := &AsyncKafkaProducer{}
	asyncProducer.Name = name
	asyncProducer.Hosts = hosts
	asyncProducer.Status = KafkaProducerDisConnected
	if config == nil {
		config = getDefaultProducerConfig(name)
	}
	asyncProducer.Config = config

	if producer, err := sarama.NewAsyncProducer(hosts, config); err != nil {
		return errwrap.Get(err, "sarama.NewAsyncProducer.error :"+name)
	} else {

		asyncProducer.Breaker = breaker.New(3, 1, 5*time.Second)
		asyncProducer.Reconnect = make(chan bool)
		asyncProducer.AsyncProducer = &producer
		asyncProducer.Status = KafkaProducerConnected
		KafkaStdLogger.Println("AsyncKakfaProducer  connected name ", name)
	}

	go asyncProducer.keepConnect()
	go asyncProducer.check()

	kafkaAsyncProducers[name] = asyncProducer
	return nil
}

// GetKafkaSyncProducer 获取kafka的同步生产者
func GetKafkaSyncProducer(name string) *SyncKafkaProducer {
	if producer, ok := kafkaSyncProducers[name]; ok {
		return producer
	} else {
		KafkaStdLogger.Println("InitSyncKafkaProducer must be called !")
		return nil
	}
}

// GetKafkaAsyncProducer 获取kafka的异步生产者
func GetKafkaAsyncProducer(name string) *AsyncKafkaProducer {
	if producer, ok := kafkaAsyncProducers[name]; ok {
		return producer
	} else {
		KafkaStdLogger.Println("InitAsyncKafkaProducer must be called !")
		return nil
	}
}

// 检查同步生产者的连接状态,如果断开链接则尝试重连
func (syncProducer *SyncKafkaProducer) keepConnect() {
	defer func() {
		KafkaStdLogger.Println("syncProducer keepConnect exited")
	}()

	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGHUP, syscall.SIGTERM, syscall.SIGQUIT, syscall.SIGINT)

	for {
		if syncProducer.Status == KafkaProducerClosed {
			return
		}

		select {
		case <-signals:
			syncProducer.StatusLock.Lock()
			syncProducer.Status = KafkaProducerClosed
			syncProducer.StatusLock.Unlock()
			return
		case <-syncProducer.Reconnect:
			if syncProducer.Status != KafkaProducerDisConnected {
				break
			}
			KafkaStdLogger.Println("kafka syncProducer ReConnecting... name " + syncProducer.Name)
			var producer sarama.SyncProducer
			for {
				if syncProducer.Status == KafkaProducerConnected {
					continue
				}
				// 利用熔断器给集群以恢复时间，避免不断的发送重联
				err := syncProducer.Breaker.Run(func() (err error) {
					producer, err = sarama.NewSyncProducer(syncProducer.Hosts, syncProducer.Config)
					return
				})

				switch err {
				case nil:
					syncProducer.StatusLock.Lock()
					if syncProducer.Status == KafkaProducerDisConnected {
						syncProducer.SyncProducer = &producer
						syncProducer.Status = KafkaProducerConnected
					}
					syncProducer.StatusLock.Unlock()
					KafkaStdLogger.Println("kafka syncProducer ReConnected, name:", syncProducer.Name)
					break
				case breaker.ErrBreakerOpen:
					KafkaStdLogger.Println("kafka connect fail, broker is open")
					//2s后重连，此时breaker刚好 half close
					if syncProducer.Status == KafkaProducerDisConnected {
						time.AfterFunc(2*time.Second, func() {
							KafkaStdLogger.Println("kafka begin to ReConnect ,because of  ErrBreakerOpen ")
							syncProducer.Reconnect <- true
						})
					}
					break
				default:
					KafkaStdLogger.Println("kafka ReConnect error, name:", syncProducer.Name, err)
				}
			}
		}
	}
}

//同步生产者状态检测
func (syncProducer *SyncKafkaProducer) check() {
	defer func() {
		KafkaStdLogger.Println("syncProducer check exited")
	}()
	// Trap SIGINT to trigger a shutdown.
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

	for {
		if syncProducer.Status == KafkaProducerClosed {
			return
		}
		select {
		case <-signals:
			syncProducer.StatusLock.Lock()
			syncProducer.Status = KafkaProducerClosed
			syncProducer.StatusLock.Unlock()
			return
		}
	}
}

// SendMessage 同步发送消息到 kafka, 单条消息
func (syncProducer *SyncKafkaProducer) SendMessage(message *sarama.ProducerMessage) (partition int32, offset int64, err error) {
	if syncProducer.Status != KafkaProducerConnected {
		return -1, -1, errors.New("kafka syncProducer " + syncProducer.Status)
	}
	partition, offset, err = (*syncProducer.SyncProducer).SendMessage(message)
	if err == nil {
		return
	}
	if errors.Is(err, sarama.ErrBrokerNotAvailable) {
		syncProducer.StatusLock.Lock()
		if syncProducer.Status == KafkaProducerConnected {
			syncProducer.Status = KafkaProducerDisConnected
			syncProducer.Reconnect <- true
		}
		syncProducer.StatusLock.Unlock()
	}
	return
}

//SendMessages 同步发送消息到 kafka 多条消息
func (syncProducer *SyncKafkaProducer) SendMessages(mses []*sarama.ProducerMessage) (errs sarama.ProducerErrors) {
	if syncProducer.Status != KafkaProducerConnected {
		return append(errs, &sarama.ProducerError{Err: errors.New("kafka syncProducer " + syncProducer.Status)})
	}
	errs = (*syncProducer.SyncProducer).SendMessages(mses).(sarama.ProducerErrors)
	for _, err := range errs {
		//触发重连
		if errors.Is(err, sarama.ErrBrokerNotAvailable) {
			syncProducer.StatusLock.Lock()
			if syncProducer.Status == KafkaProducerConnected {
				syncProducer.Status = KafkaProducerDisConnected
				syncProducer.Reconnect <- true
			}
			syncProducer.StatusLock.Unlock()
		}
	}
	return
}

//Close 关闭kafka的同步连接客户端
func (syncProducer *SyncKafkaProducer) Close() (err error) {
	syncProducer.StatusLock.Lock()
	defer syncProducer.StatusLock.Unlock()
	err = (*syncProducer.SyncProducer).Close()
	syncProducer.Status = KafkaProducerClosed
	return
}

//检查kafka连接状态,如果断开链接则尝试重连
func (asyncProducer *AsyncKafkaProducer) keepConnect() {
	defer func() {
		KafkaStdLogger.Println("asyncProducer keepConnect exited")
	}()
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)
	for {
		if asyncProducer.Status == KafkaProducerClosed {
			return
		}
		select {
		case s := <-signals:
			KafkaStdLogger.Println("kafka async producer receive system signal:" + s.String() + "; name:" + asyncProducer.Name)
			asyncProducer.Status = KafkaProducerClosed
			return
		case <-asyncProducer.Reconnect:
			if asyncProducer.Status != KafkaProducerDisConnected {
				break
			}

			KafkaStdLogger.Println("kafka syncProducer ReConnecting... name" + asyncProducer.Name)
			var producer sarama.AsyncProducer

			for {
				if asyncProducer.Status == KafkaProducerConnected {
					continue
				}
				//利用熔断器给集群以恢复时间，避免不断的发送重联
				err := asyncProducer.Breaker.Run(func() (err error) {
					producer, err = sarama.NewAsyncProducer(asyncProducer.Hosts, asyncProducer.Config)
					return
				})

				switch err {
				case nil:
					asyncProducer.StatusLock.Lock()
					if asyncProducer.Status == KafkaProducerDisConnected {
						asyncProducer.AsyncProducer = &producer
						asyncProducer.Status = KafkaProducerConnected
					}
					asyncProducer.StatusLock.Unlock()
					KafkaStdLogger.Println("kafka syncProducer ReConnected, name:" + asyncProducer.Name)
					break
				case breaker.ErrBreakerOpen:
					KafkaStdLogger.Println("kafka connect fail, broker is open")
					//2s后重连，此时breaker刚好 half close
					if asyncProducer.Status == KafkaProducerDisConnected {
						time.AfterFunc(2*time.Second, func() {
							KafkaStdLogger.Println("kafka begin to ReConnect ,because of  ErrBreakerOpen ")
							asyncProducer.Reconnect <- true
						})
					}
					break
				default:
					KafkaStdLogger.Println("kafka ReConnect error, name:"+asyncProducer.Name, zap.Error(err))
				}
			}
		}
	}
}

//异步生产者状态检测
func (asyncProducer *AsyncKafkaProducer) check() {
	defer func() {
		KafkaStdLogger.Println("asyncProducer check exited")
	}()
	for {
		switch asyncProducer.Status {
		case KafkaProducerDisConnected:
			time.Sleep(time.Second * 5)
			continue
		case KafkaProducerClosed:
			return
		}
		// Trap SIGINT to trigger a shutdown.
		signals := make(chan os.Signal, 1)
		signal.Notify(signals, syscall.SIGHUP, syscall.SIGINT, syscall.SIGTERM, syscall.SIGQUIT)

		for {
			select {
			case msg := <-(*asyncProducer.AsyncProducer).Successes():
				KafkaStdLogger.Println("Success produce message  ", zap.Any(msg.Topic, msg.Value))
			case err := <-(*asyncProducer.AsyncProducer).Errors():
				KafkaStdLogger.Println("message send error", zap.Error(err))
				if errors.Is(err, sarama.ErrOutOfBrokers) || errors.Is(err, sarama.ErrNotConnected) {
					// 连接中断触发重连，捕捉不到 EOF
					asyncProducer.StatusLock.Lock()
					if asyncProducer.Status == KafkaProducerConnected {
						asyncProducer.Status = KafkaProducerDisConnected
						asyncProducer.Reconnect <- true
					}
					asyncProducer.StatusLock.Unlock()
				}
			case s := <-signals:
				KafkaStdLogger.Println("kafka async producer receive system signal:" + s.String() + "; name:" + asyncProducer.Name)
				asyncProducer.Status = KafkaProducerClosed
				return
			}
		}
	}
}

//SendMessage 异步发送消息到 kafka 单条消息
func (asyncProducer *AsyncKafkaProducer) SendMessage(message *sarama.ProducerMessage) error {
	var err error
	if asyncProducer.Status != KafkaProducerConnected {
		return errors.New("kafka disconnected")
	}
	(*asyncProducer.AsyncProducer).Input() <- message
	return err
}

//SendMessages 异步发送消息到 kafka 多条消息
func (asyncProducer *AsyncKafkaProducer) SendMessages(messages []*sarama.ProducerMessage) error {
	var err error
	if asyncProducer.Status != KafkaProducerConnected {
		return errors.New("kafka disconnected")
	}
	for _, message := range messages {
		(*asyncProducer.AsyncProducer).Input() <- message
	}
	return err
}

//Close 关闭kafka的异步连接客户端
func (asyncProducer *AsyncKafkaProducer) Close() (err error) {
	asyncProducer.StatusLock.Lock()
	defer asyncProducer.StatusLock.Unlock()
	err = (*asyncProducer.AsyncProducer).Close()
	asyncProducer.Status = KafkaProducerClosed
	return
}
